package com.we.risk.registerchanneldeprecated.browseinfo.deprecated;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.we.flink.utils.WeKafkaPropertyReader;
import com.we.utils.CommonTimeTools;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.*;

@Deprecated
public class AdmRegisterChannelUserLabel {
    public static final String INPUT_KEY_BY = "uid";
    public static final String OUTPUT_KEY_BY = "user_key";
    public static final String LABEL_NAME = "label_name";
    public static final String LABEL_VALUE = "label_value";
    public static final String LABEL_OUTPUT_TIME = "label_output_time";

    public static final String AUDIT_TIME = "audit_time"; // +first_audit_success_time
    public static final String FIRST_ADT_SUCCESS_TIME = "first_audit_success_time";
    public static final String ID_COMBINE = "id_combine";
    public static final String HAS_RESET_XXL = "has_reset_xinxiliu";
    public static final String RECV_TIME = "receive_time"; // +unique_channel
    public static final String UNIQUE_CHANNEL = "unique_channel";
    public static final String REG_TIME = "register_time"; // +user_key
    public static final String ORIGIN_NAME = "origin_name";
    public static final String UID = "uid";

    public static final String DATEFORMAT = "yyyy-MM-dd";

    public static final String RELEASEPROP =
            "risk/registerchannel/kfk_register_channel_user_label_prod.properties";
    public static final String USERLABELPROP =
            "risk/registerchannel/adm_register_channel_user_label_prod.properties";
    public static Logger LOG = LoggerFactory.getLogger(AdmRegisterChannelUserLabel.class);

    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();
            String srckfkOffset = paramReader.getTmpKfkOffset();

            Properties consumProp = new Properties();
            consumProp.setProperty("bootstrap.servers", srcKafkaBootStrapServer);
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty("auto.offset.reset", srckfkOffset);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            FlinkKafkaConsumer<String> kfkSource =
                    new FlinkKafkaConsumer<String>(
                            srcKafkaTopic, new SimpleStringSchema(), consumProp);
            DataStreamSource<String> input = env.addSource(kfkSource);

            SingleOutputStreamOperator<String> out =
                    input.uid("AdmRegisterChannelUserLabel-input")
                            .keyBy(
                                    new KeySelector<String, String>() {
                                        @Override
                                        public String getKey(String value) throws Exception {
                                            return JSONObject.parseObject(value)
                                                    .getString(INPUT_KEY_BY);
                                        }
                                    })
                            .flatMap(new AdmRegChannelFlatMapFunc())
                            .filter(
                                    new FilterFunction<String>() {
                                        @Override
                                        public boolean filter(String value) throws Exception {
                                            JSONObject jsonObject = JSONObject.parseObject(value);
                                            if (jsonObject.containsKey(OUTPUT_KEY_BY)) {
                                                return (jsonObject.get(OUTPUT_KEY_BY) != null);
                                            } else {
                                                return false;
                                            }
                                        }
                                    })
                            .keyBy(
                                    new KeySelector<String, String>() {
                                        @Override
                                        public String getKey(String value) throws Exception {
                                            return JSONObject.parseObject(value)
                                                    .getString(OUTPUT_KEY_BY);
                                        }
                                    })
                            .flatMap(new UserLabelRichFlatMapFunc());

            /** sink to Kafka */
            String sinkkfkTopic = paramReader.getKfkTopic();
            Properties sinkProp = new Properties();
            sinkProp.setProperty("bootstrap.servers", paramReader.getKfkBootStrapServer());
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            FlinkKafkaProducer<String> userLabelAllFlinkKafkaProducer =
                    new FlinkKafkaProducer<String>(
                            sinkkfkTopic,
                            new WeKafkaKeyedSerializationSchema(),
                            sinkProp,
                            java.util.Optional.of(new WeKafkaCustomPartitioner()));

            out.addSink(userLabelAllFlinkKafkaProducer).setParallelism(sinkkfkPartitions);

            env.execute(AdmRegisterChannelUserLabel.class.toString());
        } catch (Exception e) {
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class AdmRegChannelFlatMapFunc extends RichFlatMapFunction<String, String> {
        ValueState<JSONObject> regChannelVState;
        MapState<Long, Integer> auditTimeMState;
        MapState<Long, String> recvTimeMState;
        SimpleDateFormat sdf;

        @Override
        public void open(Configuration parameters) throws Exception {
            StateTtlConfig ttlConfig =
                    StateTtlConfig.newBuilder(Time.hours(24))
                            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                            .cleanupInRocksdbCompactFilter(100000L)
                            .build();

            regChannelVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<JSONObject>(
                                            "regChannelVState", JSONObject.class));

            MapStateDescriptor<Long, Integer> auditMapDesc =
                    new MapStateDescriptor<>("auditTimeMState", Long.class, Integer.class);
            auditMapDesc.enableTimeToLive(ttlConfig);
            auditTimeMState = getRuntimeContext().getMapState(auditMapDesc);

            MapStateDescriptor<Long, String> recvmapDesc =
                    new MapStateDescriptor<>("recvTimeMState", Long.class, String.class);
            recvmapDesc.enableTimeToLive(ttlConfig);
            recvTimeMState = getRuntimeContext().getMapState(recvmapDesc);
            sdf = new SimpleDateFormat(DATEFORMAT);
        }

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            try {
                // LOG.warn("input: " + value);
                JSONObject inputJson = JSONObject.parseObject(value);
                if (inputJson != null) {
                    JSONObject tmpState = regChannelVState.value();
                    JSONObject curState =
                            (tmpState == null ? initJsonObjState(USERLABELPROP) : tmpState);

                    Long stateLastAdtTime = curState.getLong(AUDIT_TIME);
                    Integer stateHasResetXxl = curState.getInteger(HAS_RESET_XXL);
                    Long stateRcvTime = curState.getLong(RECV_TIME);

                    if (inputJson.containsKey(AUDIT_TIME)) {
                        // audit列表 + 更新初次成功进件时间
                        long auditTime = inputJson.getLong(AUDIT_TIME);

                        if (stateHasResetXxl == null) {
                            auditTimeMState.put(auditTime, 1);
                        }

                        if (stateLastAdtTime == null || auditTime > stateLastAdtTime) {
                            curState.put(AUDIT_TIME, auditTime);
                        }
                        // 首次成功进件时间
                        if (inputJson.containsKey(FIRST_ADT_SUCCESS_TIME)) {
                            curState.put(
                                    FIRST_ADT_SUCCESS_TIME, inputJson.get(FIRST_ADT_SUCCESS_TIME));
                        }
                    }
                    if (inputJson.containsKey(RECV_TIME)) {
                        long recvTime = inputJson.getLongValue(RECV_TIME);
                        String channel = inputJson.getString(UNIQUE_CHANNEL);
                        // 广告点击列表
                        if (stateHasResetXxl == null) {
                            recvTimeMState.put(recvTime, channel);
                        }
                        // 最近一次点击时间+channel
                        if (stateRcvTime == null || recvTime > stateRcvTime) {
                            curState.put(RECV_TIME, recvTime);
                            curState.put(UNIQUE_CHANNEL, channel);
                        }
                    }
                    if (inputJson.containsKey(REG_TIME)) {
                        curState.put(REG_TIME, inputJson.get(REG_TIME));
                        curState.put(OUTPUT_KEY_BY, inputJson.get(OUTPUT_KEY_BY));
                        curState.put(UID, inputJson.get(UID));
                    }
                    if (inputJson.containsKey(ID_COMBINE)) {
                        curState.put(ID_COMBINE, inputJson.get(ID_COMBINE));
                    }
                    if (inputJson.containsKey(ORIGIN_NAME)) {
                        curState.put(ORIGIN_NAME, inputJson.get(ORIGIN_NAME));
                    }

                    // 修改注册归因流程
                    // LOG.warn("stateHasResetXxl: " + (stateHasResetXxl == null ? "null" :
                    // stateHasResetXxl));
                    if (stateHasResetXxl == null) {
                        Long regTime = curState.getLong(REG_TIME);
                        Long firstAdtSucTime = curState.getLong(FIRST_ADT_SUCCESS_TIME);
                        Long lastAdtTime = curState.getLong(AUDIT_TIME);
                        Long lastRecvTime = curState.getLong(RECV_TIME);
                        // LOG.warn("regTime: " + (regTime == null ? "null" : regTime));
                        if (regTime != null) {
                            // 状态ttl未过期
                            // LOG.warn("auditTimeMState empty: " + auditTimeMState.isEmpty() + "
                            // recvTimeMState empty: " + recvTimeMState.isEmpty());
                            if (!auditTimeMState.isEmpty() && !recvTimeMState.isEmpty()) {
                                Iterator<Long> auditTimeList = auditTimeMState.keys().iterator();
                                Iterator<Long> recvTimeList = recvTimeMState.keys().iterator();

                                if (auditTimeList.hasNext() && recvTimeList.hasNext()) {
                                    while (auditTimeList.hasNext()) {
                                        Long auditTime = auditTimeList.next();
                                        // 进件时间>注册时间 30天以上
                                        // LOG.warn("#1auditTime: " + auditTime + " regTime: " +
                                        // regTime + " datediff: " +
                                        // CommonTimeTools.dateDiffBetAuditRegTime(auditTime,
                                        // regTime, sdf));
                                        if (CommonTimeTools.dateDiffBetAuditRegTime(
                                                        auditTime, regTime, sdf)
                                                > 30) {
                                            //
                                            // Iterator<Long> recvTimeList =
                                            // recvTimeMState.keys().iterator();
                                            while (recvTimeList.hasNext()) {
                                                Long recvTime = recvTimeList.next();
                                                // LOG.warn("#1recvTime > regTime: " + (recvTime >
                                                // regTime) + " recvTime < auditTime: " + (recvTime
                                                // < auditTime));
                                                // 点击时间在注册时间和进件时间范围内，并且是首贷用户
                                                if (recvTime > regTime && recvTime < auditTime) {
                                                    // 避免乱序造成计算差错，要求audit表先上线，点击流最后上线
                                                    if (firstAdtSucTime == null
                                                            || recvTime < firstAdtSucTime) {
                                                        // 打标 信息流用户
                                                        stateHasResetXxl = 1;
                                                        curState.put(
                                                                HAS_RESET_XXL, stateHasResetXxl);
                                                        // LOG.warn("return outJson: " +
                                                        // curState.toString());
                                                        out.collect(curState.toString());
                                                        regChannelVState.update(curState);
                                                        return;
                                                    }
                                                }
                                            }
                                        }
                                    }
                                } else if (auditTimeList.hasNext()) {
                                    // LOG.warn("#2lastRecvTime: " + (lastRecvTime == null ? "null"
                                    // : lastRecvTime));
                                    if (lastRecvTime != null) {
                                        while (auditTimeList.hasNext()) {
                                            Long auditTime = auditTimeList.next();
                                            // LOG.warn("#2auditTime: " + auditTime + " regTime: " +
                                            // regTime + " datediff: " +
                                            // CommonTimeTools.dateDiffBetAuditRegTime(auditTime,
                                            // regTime, sdf));
                                            if (CommonTimeTools.dateDiffBetAuditRegTime(
                                                            auditTime, regTime, sdf)
                                                    > 30) {
                                                // LOG.warn("#2recvTime > regTime: " + (lastRecvTime
                                                // > regTime) + " recvTime < auditTime: " +
                                                // (lastRecvTime < auditTime));
                                                if (lastRecvTime > regTime
                                                        && lastRecvTime < auditTime) {
                                                    if (firstAdtSucTime == null
                                                            || lastRecvTime < firstAdtSucTime) {
                                                        stateHasResetXxl = 1;
                                                        curState.put(
                                                                HAS_RESET_XXL, stateHasResetXxl);
                                                        break;
                                                    }
                                                }
                                            }
                                        }
                                    }

                                } else if (recvTimeList.hasNext()) {
                                    // LOG.warn("#3lastAdtTime: " + (lastAdtTime == null ? "null" :
                                    // lastAdtTime));
                                    if (lastAdtTime != null) {
                                        while (recvTimeList.hasNext()) {
                                            Long recvTime = recvTimeList.next();
                                            // LOG.warn("#3 lastAdtTime: " + lastAdtTime + "
                                            // regTime: " + regTime + " datediff: " +
                                            // CommonTimeTools.dateDiffBetAuditRegTime(lastAdtTime,
                                            // regTime, sdf));
                                            if (CommonTimeTools.dateDiffBetAuditRegTime(
                                                            lastAdtTime, regTime, sdf)
                                                    > 30) {
                                                // LOG.warn("#3recvTime > regTime: " + (recvTime >
                                                // regTime) + " recvTime < auditTime: " + (recvTime
                                                // < lastAdtTime));
                                                if (recvTime > regTime && recvTime < lastAdtTime) {
                                                    if (firstAdtSucTime == null
                                                            || recvTime < firstAdtSucTime) {
                                                        stateHasResetXxl = 1;
                                                        curState.put(
                                                                HAS_RESET_XXL, stateHasResetXxl);
                                                        break;
                                                    }
                                                }
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                    // LOG.warn("final outJson: " + curState.toString());
                    if (curState.get(OUTPUT_KEY_BY) != null) {
                        out.collect(curState.toString());
                    }
                    // LOG.warn("==============================================================");

                    regChannelVState.update(curState);
                }
            } catch (Exception e) {
                LOG.error("Excep: " + e.toString());
                LOG.error("Excep input: " + value);
            }
        }
    }

    private static class UserLabelRichFlatMapFunc extends RichFlatMapFunction<String, String> {
        ValueState<JSONObject> userLabelVState;

        @Override
        public void open(Configuration parameters) throws Exception {
            userLabelVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<JSONObject>(
                                            "userLabelVState", JSONObject.class));
        }

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            JSONObject tmp = userLabelVState.value();
            JSONObject curState = (tmp == null ? initJsonObjState(USERLABELPROP) : tmp);
            JSONObject input = JSON.parseObject(value);
            String userKeyVal = input.getString(OUTPUT_KEY_BY);
            for (Map.Entry<String, Object> inputEle : input.entrySet()) {
                String key = inputEle.getKey();
                Object val = inputEle.getValue();

                if (!key.equals(OUTPUT_KEY_BY) && !key.equals(INPUT_KEY_BY)) {
                    Object jsonVal = curState.get(key);
                    if (jsonVal != null && val != null) {
                        if (!jsonVal.equals(val)) {
                            curState.put(key, val);
                            JSONObject jsonObject = new JSONObject();
                            jsonObject.put(OUTPUT_KEY_BY, userKeyVal);
                            jsonObject.put(LABEL_NAME, key);
                            jsonObject.put(LABEL_VALUE, val);
                            jsonObject.put(LABEL_OUTPUT_TIME, System.currentTimeMillis());
                            out.collect(jsonObject.toString());
                        }
                    } else {
                        if (jsonVal != val) {
                            curState.put(key, val);
                            JSONObject jsonObject = new JSONObject();
                            jsonObject.put(OUTPUT_KEY_BY, userKeyVal);
                            jsonObject.put(LABEL_NAME, key);
                            jsonObject.put(LABEL_VALUE, val);
                            jsonObject.put(LABEL_OUTPUT_TIME, System.currentTimeMillis());
                            out.collect(jsonObject.toString());
                        }
                    }
                }
            }
            userLabelVState.update(curState);
        }
    }

    private static JSONObject initJsonObjState(String propFile) throws IOException {

        JSONObject res = new JSONObject();
        InputStream resourceAsStream =
                AdmRegisterChannelUserLabel.class.getClassLoader().getResourceAsStream(propFile);
        ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
        Set<Object> propSets = reader.getProperties().keySet();
        for (Object key : propSets) {
            String typeValue = reader.get((String) key);
            String[] splits = typeValue.split(":");
            if (splits.length == 2) {
                String type = splits[0];
                String value = splits[1];
                jsonObjectSetValue(res, type, (String) key, value);
            } else {
                LOG.error("split Error: " + splits.length);
            }
        }
        return res;
    }

    private static void jsonObjectSetValue(
            JSONObject jsonObj, String type, String key, String value) {
        if (!value.equals("null")) {
            switch (type) {
                case "Long":
                    jsonObj.put(key, Long.parseLong(value));
                    break;
                case "String":
                    jsonObj.put(key, value);
                    break;
                case "Float":
                    jsonObj.put(key, Float.parseFloat(value));
                    break;
                case "Integer":
                    jsonObj.put(key, Integer.parseInt(value));
                    break;
                default:
                    LOG.error("Not support this type: " + type);
                    break;
            }
        } else {
            jsonObj.put(key, null);
        }
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String key = jsonObject.getString(OUTPUT_KEY_BY);
            return key.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }
}
